package com.ksvip.next.components.redis;

import java.io.IOException;
import java.util.concurrent.Callable;

import com.ksvip.next.components.core.bean.BaseDTO;
import com.ksvip.next.components.redis.constants.RedisConstant;
import com.ksvip.next.components.redis.retry.RedisRetryerBuilder;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.Validate;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.github.rholder.retry.Retryer;

import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPoolConfig;
import redis.clients.jedis.exceptions.JedisConnectionException;

public class JedisPoolManager {
    private final static Logger log           = LoggerFactory.getLogger(JedisPoolManager.class);
    /**
     * redis相关配置
     */
    private int                 maxTotal      = RedisConstant.POOL_MAX_TOTAL;
    private int                 maxIdle       = RedisConstant.POOL_MAX_IDLE;
    private int                 timeOut       = RedisConstant.DEFAULT_TIMEOUT;
    private long                maxWaitMillis = -1;
    private String              host;
    private String              password;
    private int                 port;
    private int                 dbNo;
    private JedisPool           pool;
    private final String        version       = "1.0.0";

    /**
     * 环境 默认test
     */
    private String              env           = RedisConstant.DEV_ENV;
    /**
     * zk 跟节点
     */
    private final String        zkRoot        = RedisConstant.ZK_ROOT;
    /**
     * zk 子节点(根据公司部门等相关信息生成)
     */
    private String              redisConfigKey;
    /**
     * zk 服务地址
     */
    private String              zkServer;

    public JedisPoolManager() {
        super();
    }

    public JedisPoolManager(String host, String password, int port) {
        super();
        this.host = host;
        this.password = password;
        this.port = port;
    }

    public JedisPoolManager(String redisConfigKey, String zkServer, int maxTotal, int maxIdle, int timeOut) {
        super();
        this.maxTotal = maxTotal;
        this.maxIdle = maxIdle;
        this.timeOut = timeOut;
        this.redisConfigKey = redisConfigKey;
        this.zkServer = zkServer;
    }

    public JedisPoolManager(String redisConfigKey, String zkServer, int maxTotal, int maxIdle, int timeOut,
                            int maxWaitMillis) {
        super();
        this.maxTotal = maxTotal;
        this.maxIdle = maxIdle;
        this.timeOut = timeOut;
        this.redisConfigKey = redisConfigKey;
        this.zkServer = zkServer;
        this.maxWaitMillis = maxWaitMillis;
    }

    public void init() {
        JedisPoolConfig config = new JedisPoolConfig();
        // 最大空闲连接数, 应用自己评估，不要超过ApsaraDB for Redis每个实例最大的连接数
        config.setMaxIdle(this.maxIdle);
        // 最大连接数, 应用自己评估，不要超过ApsaraDB for Redis每个实例最大的连接数
        config.setMaxTotal(this.maxTotal);
        config.setTestOnBorrow(true);//向调用者输出“链接”资源时，是否检测是有有效，如果无效则从连接池中移除，并尝试获取继续获取。设为true
        config.setTestOnReturn(true);//向连接池“归还”链接时，是否检测“链接”对象的有效性
        config.setMaxWaitMillis(this.maxWaitMillis);
        try {
            if (StringUtils.isEmpty(this.host) || StringUtils.isEmpty(this.password)) {
                this.pool = this.creatJedisPoolByZk(config, this.redisConfigKey, this.zkServer);
            } else {
                this.pool = new JedisPool(config, this.host, this.port, this.timeOut, this.password);
                log.info(
                        "========> jedis pool is created  by  JedisPool,version:{}, host:{},port:{},dbNo:{},maxIdle:{},maxtotal:{},timeout:{} maxWaitMillis:{}",
                        this.version, this.host, this.port, this.dbNo, this.maxIdle, this.maxTotal, this.timeOut,
                        this.maxWaitMillis);
            }
        } catch (Exception e) {
            log.error("jedis pool is created fail " + e.getMessage(), e);
            throw new RuntimeException("jedis pool is created fail " + e.getMessage(), e);
        }

    }

    /**
     * 根据redis db 获取jedis
     *
     * @return
     */
    public Jedis jedis() {
        Jedis jedis = null;
        try {
            jedis = this.pool.getResource();
        } catch (JedisConnectionException e) {
            log.error(e.getMessage(), e);
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        }
        return jedis;
    }

    /**
     * 获取jedis
     */
    public Jedis getJedis() {
        try {
            final Retryer<Jedis> retryer = RedisRetryerBuilder.build();
            final Jedis jedis = retryer.call(new Callable<Jedis>() {
                @Override
                public Jedis call() throws Exception {
                    return JedisPoolManager.this.jedis();
                }
            });
            return jedis;
        } catch (final Exception e) {
            log.error("多次获取Redis连接失败！");
        }
        Validate.validState(false, "pool属性未注入, 请在applicationContext.xml中定义JedisCache.");
        return null;
    }

    /**
     * 关闭jedis
     *
     * @param jedis
     */
    public void close(Jedis jedis) {
        if (jedis != null) {
            jedis.close();
        }
    }

    /**
     * 销毁链接池
     */
    public void destory() {
        if (this.pool != null) {
            log.info("====> jedisPool is destory ");
            this.pool.destroy();
        }
    }

    private JedisPool creatJedisPoolByZk(JedisPoolConfig config, String redisConfigKey, String zkServer) {
        ZooKeeper zooKeeper = null;
        String data = null;
        int count = 0;
        while (count <= 3) {
            try {
                zooKeeper = this.getZooKeeper(zkServer);
                data = new String(zooKeeper.getData(redisConfigKey, false, null));
                log.info("zooKeeper:{} is Available", zkServer);
                break;
            } catch (Exception e) {
                count++;
                if (count == 3) {
                    throw new RuntimeException("Cannot connect to zooKeeper running " + e.getMessage(), e);
                }
                log.warn("Cannot connect to zooKeeper running @ " + zkServer + ". Trying next one.");
            }
        }

        RedisInfo redisInfo = BaseDTO.fromJson(data, RedisInfo.class);
        this.pool = new JedisPool(config, redisInfo.getHost(), redisInfo.getPort(), this.timeOut,
                redisInfo.getPassword(), redisInfo.getDbNo());
        this.setDbNo(redisInfo.getDbNo());
        this.setHost(redisInfo.getHost());
        this.setPort(redisInfo.getPort());
        this.setZkServer(zkServer);
        log.info(
                "=========>jedis pool is created  by  zk,version:{},host:{},port:{},dbNo:{},maxIdle:{},maxtotal:{},timeout:{} maxWaitMillis:{}",
                this.version, redisInfo.getHost(), redisInfo.getPort(), redisInfo.getDbNo(), this.maxIdle,
                this.maxTotal, this.timeOut, this.maxWaitMillis);
        try {
            this.closeZookeeper(zooKeeper);
        } catch (InterruptedException e) {
            log.warn("close zookeeper error." + e.getMessage(), e);
        }
        return this.pool;

    }

    private ZooKeeper getZooKeeper(String server) throws IOException {
        ZooKeeper zk = new ZooKeeper(server, 1000 * 20, new Watcher() {
            // 监控所有被触发的事件
            @Override
            public void process(WatchedEvent event) {
                log.info(event.toString());
            }
        });
        return zk;
    }

    private void closeZookeeper(ZooKeeper zk) throws InterruptedException {
        if (zk != null) {
            zk.close();
        }

    }

    public void setMaxTotal(int maxTotal) {
        this.maxTotal = maxTotal;
    }

    public void setMaxIdle(int maxIdle) {
        this.maxIdle = maxIdle;
    }

    public void setTimeOut(int timeOut) {
        this.timeOut = timeOut;
    }

    public void setHost(String host) {
        this.host = host;
    }

    public void setPassword(String password) {
        this.password = password;
    }

    public void setPort(int port) {
        this.port = port;
    }

    public void setDbNo(int dbNo) {
        this.dbNo = dbNo;
    }

    public void setPool(JedisPool pool) {
        this.pool = pool;
    }

    public void setRedisConfigKey(String redisConfigKey) {
        this.redisConfigKey = redisConfigKey;
    }

    public void setZkServer(String zkServer) {
        this.zkServer = zkServer;
    }

    public int getMaxTotal() {
        return this.maxTotal;
    }

    public int getMaxIdle() {
        return this.maxIdle;
    }

    public int getTimeOut() {
        return this.timeOut;
    }

    public String getHost() {
        return this.host;
    }

    public String getPassword() {
        return this.password;
    }

    public int getPort() {
        return this.port;
    }

    public int getDbNo() {
        return this.dbNo;
    }

    public JedisPool getPool() {
        return this.pool;
    }

    public String getVersion() {
        return this.version;
    }

    public String getZkRoot() {
        return this.zkRoot;
    }

    public String getRedisConfigKey() {
        return this.redisConfigKey;
    }

    public String getZkServer() {
        return this.zkServer;
    }

    public long getMaxWaitMillis() {
        return this.maxWaitMillis;
    }

    public void setMaxWaitMillis(long maxWaitMillis) {
        this.maxWaitMillis = maxWaitMillis;
    }

    public String getEnv() {
        return this.env;
    }

    public void setEnv(String env) {
        this.env = env;
    }

}
